package com.czl.myRabbitMq.mqClient;


import com.czl.myRabbitMq.commom.BinaryTransformTool;
import com.czl.myRabbitMq.commom.Request;
import com.czl.myRabbitMq.commom.Response;
import com.czl.myRabbitMq.commom.exception.mqException;
import com.czl.myRabbitMq.commom.returns.DescribeReturns;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class Connection {
    private Socket socket = null;
    //这里在一个连接中需要用channelMap对象来管理多个Channel对象
    private ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>();
    private InputStream inputStream;
    private OutputStream outputStream;
    private DataInputStream dataInputStream;
    private DataOutputStream dataOutputStream;
    private ExecutorService executorService;
    public Connection(String ip, int port) throws IOException {
        socket = new Socket(ip, port);

        this.inputStream = socket.getInputStream();
        this.outputStream = socket.getOutputStream();
        this.dataInputStream = new DataInputStream(inputStream);
        this.dataOutputStream = new DataOutputStream(outputStream);
        executorService = Executors.newFixedThreadPool(5);

        //同时需要一个扫描线程 不断扫描接受客户端传来的消息
        Thread t = new Thread(()->{
           while (!socket.isClosed()){
               try {
                   Response response = getResponse();
                   handleResponse(response);
               }catch (SocketException e){
                   log.info("连接正常断开");
               } catch (IOException | mqException e) {
                   log.info("连接出现异常");
                   throw new RuntimeException(e);
               }
           }
        });
    }

    public void close(){
        try {
            executorService.shutdownNow();
            channelMap.clear();
            inputStream.close();
            outputStream.close();
            socket.close();
        }catch (IOException e){
            e.printStackTrace();
        }
        }

    private void handleResponse(Response response) throws mqException {
        //表示当前返回对象是一条消息
        if(response.getType() == 0xc){
            byte[] body = response.getBody();
            DescribeReturns describeReturns = (DescribeReturns) BinaryTransformTool.fromData(body);
            //找出这个相应对应的Channel
            Channel channel = channelMap.get(describeReturns.getChannelID());
            if(channel == null){
                throw new mqException("返回消息中对应的信道不存在" + describeReturns.getChannelID());
            }
            //执行Channel中的回调
            executorService.submit(()->{
               try{
                   channel.getConsume().delivery(describeReturns.getConsumerTag(), describeReturns.getBasicProperties(),
                           describeReturns.getBody());
               }catch (Exception e){
                   e.printStackTrace();
               }
            });
        }else {
            //其他消息的正常返回情况
            byte[] body = response.getBody();
            DescribeReturns describeReturns = (DescribeReturns) BinaryTransformTool.fromData(body);
            //将这个结果放到对应channel的哈希表中
            Channel channel = channelMap.get(describeReturns.getChannelID());
            if(channel == null){
                throw new mqException("返回消息中对应的信道不存在" + describeReturns.getChannelID());
            }
            channel.sendMessage(describeReturns);
        }
    }

    //发送一个请求给服务器
    public void writeRequest(Request request) throws IOException {
        dataOutputStream.writeInt(request.getType());
        dataOutputStream.writeInt(request.getLength());
        dataOutputStream.write(request.getBody());
        log.info("接收到请求" + request.getType());
        dataOutputStream.flush();
    }
    //接收一个请求
    public Response getResponse() throws IOException, mqException {
        Response response = new Response();
        response.setType(dataInputStream.readInt());
        response.setLength(dataInputStream.readInt());
        byte[] body = new byte[response.getLength()];
        int actual = dataInputStream.read(body);
        if(actual != body.length){
            throw new mqException("传输的数据不完整");
        }
        response.setBody(body);
        log.info("请求读取成功");
        return response;
    }
    //创建出一个Channel对象
    public Channel createChannel() throws IOException {
        String channelId = "C-" + UUID.randomUUID();
        //创建出一个Channel对象
        Channel channel = new Channel(channelId, this);
        channelMap.put(channelId, channel);
        //同时我们需要将创建一个Channel的消息发送给服务器
        //这个方法用于告诉服务器 我们创建了一个Channel
        boolean ok = channel.createChannel();
        if(!ok){
            //表示服务端创建Channel对象失败 需要将ChannelMap中的值删除
            channelMap.remove(channelId);
            log.info("服务器创建channel信道过程出现错误");
            return null;
        }

        return channel;
    }


}
